package net.bwie.realtime.newcar.dws.job;


import net.bwie.realtime.newcar.dws.function.CarChargingAnalysisFuction;
import net.bwie.realtime.newcar.utils.DorisUtil;
import net.bwie.realtime.newcar.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class CarChargingAnalysisJob {
    public static void main(String[] args) throws Exception {
        // 1.env执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        // 2.消费kafka
        DataStream<String> KafkaString = KafkaUtil.consumerKafka(env, "new_car_log_info");
//        KafkaString.print();
        // 3.数据转换
        DataStream<String> processKafkaStream = processLog(KafkaString);
        processKafkaStream.print();
        // 4.消费到kafka
        KafkaUtil.producerKafka(processKafkaStream,"new-car-charging-analysis");
        // 存入Doris
        DorisUtil.saveToDoris(
                processKafkaStream,
                "ner_car_report",
                "new_car_charging_analysis"
        );
        // 5.触发执行
        env.execute("CarChargingAnalysisJob");
    }

    private static DataStream<String> processLog(DataStream<String> stream) {
        SingleOutputStreamOperator<NewCarBean> beanStream = stream.map(new MapFunction<String, NewCarBean>() {
            @Override
            public NewCarBean map(String s) throws Exception {
                String[] str = s.split(",");
                return new NewCarBean(
                        str[0],
                        str[1],
                        str[2],
                        Double.parseDouble(str[3]),
                        Double.parseDouble(str[4]),
                        Long.parseLong(str[5]),
                        Double.parseDouble(str[6]),
                        str[7],
                        str[8],
                        Double.parseDouble(str[9]),
                        Double.parseDouble(str[10]),
                        str[11]
                );
            }
        });
//        // 设置水位线,乱序2秒
//        SingleOutputStreamOperator<NewCarBean> timeStream = beanStream.assignTimestampsAndWatermarks(
//                WatermarkStrategy
//                        .<NewCarBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
//                        .withTimestampAssigner(new SerializableTimestampAssigner<NewCarBean>() {
//                            @Override
//                            public long extractTimestamp(NewCarBean newCarBean, long l) {
//                                return newCarBean.getTimestamp();
//                            }
//                        })
//        );
        // 汽车id分组
        SingleOutputStreamOperator<NewCarBean> filterStream = beanStream.filter(new FilterFunction<NewCarBean>() {
            @Override
            public boolean filter(NewCarBean newCarBean) throws Exception {
                if (newCarBean.getChargingStatus().equals("charging")) {
                    return true;
                }
                return false;
            }
        });
        KeyedStream<NewCarBean, String> keyStream = filterStream.keyBy(bean -> bean.getVin());
//        // 设置滚动窗口,窗口大小1分钟
//        WindowedStream<NewCarBean, String, TimeWindow> windowStream = keyStream.window(
//                TumblingEventTimeWindows.of(Time.minutes(1))
//        );
        SingleOutputStreamOperator<String> processStream = keyStream.process(new CarChargingAnalysisFuction());
        return processStream;
    }
}
